#include <stdio.h>
#include <unistd.h>
#include <error.h>
#include <string.h>
#include <sstream>
#include <algorithm>

#include "../inc/rps_common.h"
#include "../inc/json/json.h"
static pthread_mutex_t test_lock_2 = PTHREAD_MUTEX_INITIALIZER;
static pthread_mutex_t g_hash_lock = PTHREAD_MUTEX_INITIALIZER;
//static pthread_mutex_t g_conn_lock = PTHREAD_MUTEX_INITIALIZER;
typedef std::unordered_map< std::string, conn* > hash_table_t;
hash_table_t hash_table; 

extern conn** conns;

void update_timer(struct event* timer, int sec) {
	if (timer != NULL) {
		struct timeval tv = { 0, 0 };
		tv.tv_sec = sec;
		event_add(timer, &tv);
	}
}

std::string get_string( const Json::Value& table )
{
	std::string tmp("");
	Json::StyledWriter writer(tmp);
	writer.write(table);
	return tmp;
}

void timer_handler(int fd, short which, void* args) {
	conn* c = (conn*)(args);
	assert(c);
	
	if (which & EV_TIMEOUT) {
		
		printf( "==============================================\n" );
		fprintf(stderr, "fd=%d timeout succ ...\n", fd);
		
		if (c->event != NULL) {
			evtcp_safe_free(c->event, event_free);
		}

		close(c->sfd);

		memset(c->r_buf, 0, c->r_size);
		memset(c->w_buf, 0, c->w_size);
		c->r_curr = c->r_buf;
		c->w_curr = c->w_buf;
		c->r_bytes = c->w_bytes = 0;
		
		delete_hash_table(c->uuid);
	}
}

int device_register(conn* c
					, std::string uuid
					, std::string area
					, std::string oemid
					, std::string dev_port) {
	int ret = 0;
	conn* node = check_hash_table(uuid);
	if (node == NULL) {
		memcpy(c->uuid, uuid.c_str(), uuid.length());
		memcpy(c->area, area.c_str(), area.length());
		memcpy(c->oemid, oemid.c_str(), oemid.length());

		insert_hash_table(uuid, c);
	} else {
		if (node != c) {
			fprintf(stderr, "aaaaaaaaaaaaaaaaaaaaaaaaaaaaa\n");
			printf( "node addr=%p\n", node );
			printf( "node->event=%p\n", node->event );
			printf( "node->sfd=%d\n", node->sfd);
			delete_hash_table(uuid);
			pthread_mutex_lock(&test_lock_2);
			if (node->event != NULL) {
				evtcp_safe_free(node->event, event_free);
			}
			pthread_mutex_unlock(&test_lock_2);
			close(node->sfd);
		} else {
		
		}
	}	
/*		
	struct timeval tv;
	gettimeofday( &tv, NULL );
	int status = c->thread->redis_conn->status;
	
	if( status == 1 && ( ( tv.tv_sec - c->flush_time ) > 5 ) )
	{
		printf( "status==1 hmset\n" );
		char cmd_hmset[256] = { '\0' };
		snprintf( cmd_hmset, sizeof( cmd_hmset )
					, "HMSET %s ServerIP %s ServerPort %d DevicePort %d"
						, c->uuid, "6.6.6.6", 6610, 3456 );
		
		redisAsyncCommand( c->thread->redis_conn->ctx
							, redis_op_status, c->thread->redis_conn
								, cmd_hmset );
								
		char cmd_expire[128] = { '\0' };
		snprintf( cmd_expire, sizeof( cmd_expire )
					, "EXPIRE %s %d", c->uuid, 30 );
		redisAsyncCommand( c->thread->redis_conn->ctx
							, redis_op_status, c->thread->redis_conn
								, cmd_expire );
		printf( "cmd_expire=%s\n", cmd_expire );
		
		c->flush_time = tv.tv_sec;
	}
*/	
	Json::Value rsptab = Json::Value::null;
	rsptab["AgentProtocol"]["Header"]["CSeq"] = "111";
	rsptab["AgentProtocol"]["Header"]["Version"] = "1.0"; 
	rsptab["AgentProtocol"]["Header"]["MessageType"] = "MSG_AGENT_REGISTER_RSP";
	rsptab["AgentProtocol"]["Header"]["ErrorNum"] = "200"; 
	rsptab["AgentProtocol"]["Header"]["ErrorString"] = "Success OK"; 
	rsptab["AgentProtocol"]["Body"]["KeepAliveIntervel"] = std::to_string( HEART_BEAT_INTERVAL );
	std::string rspstr = get_string( rsptab );
	
	int body_length = rspstr.length();
	std::string http_body_length = std::to_string( body_length );
	std::string http_request_line = "HTTP/1.1 200 OK\r\n";
	std::string http_content_head = "Content-Type: text/plain\r\n";
	std::string http_content_length = "Content-Length: ";
	std::string http_end = "\r\n\r\n";
	std::string http_rsp = http_request_line + http_content_head + http_content_length + http_body_length + http_end + rspstr;
	
	int rsplen = http_rsp.length();
	
	ret = write(c->sfd, http_rsp.c_str(), rsplen);
	if (ret != rsplen ) {
		perror("Write error");
	}
	
	return 0;
}

void insert_hash_table( std::string uuid, conn* c )
{	
	pthread_mutex_lock( &g_hash_lock );
	hash_table[uuid] = c;
//	hash_table.insert( make_pair( uuid, c ) );
	printf( "hash table size=%ld\n", hash_table.size() );
	
	pthread_mutex_unlock( &g_hash_lock );
}

void delete_hash_table( std::string uuid )
{	
	pthread_mutex_lock( &g_hash_lock );
	
	hash_table_t::iterator iter = hash_table.find(uuid);
	if (iter != hash_table.end()) {
		hash_table.erase(iter);
	}
	
	printf( "hash table size=%ld\n", hash_table.size() );
	
	pthread_mutex_unlock( &g_hash_lock );
}

//Check whether the device has been registered.
conn* check_hash_table(std::string uuid) {	
	conn* c = NULL;
	pthread_mutex_lock(&g_hash_lock);
	
	hash_table_t::iterator iter = hash_table.find(uuid);
	if(iter != hash_table.end()) {
		c = iter->second;
	}
	
	pthread_mutex_unlock(&g_hash_lock);

	return c;
}

//////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////

void redis_conn_cb(const struct redisAsyncContext* ctx, int status) {
	if(ctx == NULL || ctx->data == NULL) {
		return;
	}
	
	redis* handle = ( redis* )ctx->data;
	assert( handle != NULL );
	
	//判断这个ev_check是否经过初始化
	if( event_initialized( &handle->ev_check ) )
	{
		event_del( &handle->ev_check );
	}
		
	if( status == REDIS_OK )
	{
		printf( "Connect redis succ ...\n" );
		handle->status = 1;
		
		//event_assign第一个参数需要一个未初始化的event结构
		event_assign( &handle->ev_check, handle->base, -1, 0, redis_check_health_cb, handle );
	}
	else
	{
		printf( "Connect redis met error\n" );
		handle->status = 0;
		
		event_assign( &handle->ev_check, handle->base, -1, 0, redis_reconn_cb, handle );
	}
	
	update_timer( &handle->ev_check, 10 );
}

void redis_check_health_cb( int fd, short events, void* args )
{
	if( args == NULL )
	{
		return;
	}

	redis* handle = ( redis* )args;
	assert( handle );
	
	if( handle->status == 1 )
	{
		const char* cmd_test = "SET wang peng";
		redisAsyncCommand( handle->ctx, redis_op_status, handle, cmd_test );
	}
}

void redis_op_status( redisAsyncContext* ctx, void* reply, void* args )
{
	if( args == NULL )
	{
		return;
	}
	
	redis* handle = ( redis* )args;
	redisReply* rpy = ( redisReply* )reply;
	if( !rpy )
	{
		printf( "run redis test cmd met error\n" );
		redisAsyncDisconnect( ctx );
		return;
	}

	printf( "run redis test cmd succ...\n" );
	update_timer( &handle->ev_check, 10 );
}

void redis_disconn_cb( const struct redisAsyncContext* ctx, int status )
{
	if( ctx == NULL )
	{
		return;
	}
	
	redis* handle = ( redis* )ctx->data;
	assert( handle );
	handle->status = 0;
	
	if( event_initialized( &handle->ev_check ) )
	{
		event_del( &handle->ev_check );
	}
	
	event_assign( &handle->ev_check, handle->base, -1, 0, redis_reconn_cb, handle );
	update_timer( &handle->ev_check, 10 );
}

void redis_reconn_cb( int fd, short events, void* args )
{
	if( args == NULL )
	{
		return;
	}

	redis* handle = ( redis* )args;
	assert( handle );
	
	const char* addr = "10.10.18.2";
	const int port = 6379;
	handle->ctx = redisAsyncConnect( addr, port );
	handle->ctx->data = handle;
	
	redisLibeventAttach( handle->ctx, handle->base );
	redisAsyncSetConnectCallback( handle->ctx, redis_conn_cb );
	redisAsyncSetDisconnectCallback( handle->ctx, redis_disconn_cb );
}



